package com.shujia.flink.tf

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo2FlatMap {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.readTextFile("data/words.txt")


    //scala api
    val scalaDS: DataStream[String] = linesDS.flatMap(_.split(","))

    //java api
    val wordDS: DataStream[String] = linesDS.flatMap(new FlatMapFunction[String, String] {
      /**
        *
        * @param value ： 原始的数据类型
        * @param out   ：用于将数据发送到下游
        */
      override def flatMap(value: String, out: Collector[String]): Unit = {

        val split: Array[String] = value.split(",")

        //将数据一条一条发送到下游
        for (word <- split) {

          //将数据发送到下游
          out.collect(word)

        }
      }
    })

    wordDS.print()

    env.execute()

  }

}
